package io.jnet.client.read;

import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import com.jfireframework.baseutil.collection.buffer.ByteBuf;
import io.jnet.client.api.ClientChannelContext;
import io.jnet.client.api.ClientListener;
import io.jnet.client.api.ReadHandler;
import io.jnet.common.decodec.DecodeResult;
import io.jnet.common.decodec.FrameDecodec;
import io.jnet.common.exception.EndOfStreamException;
import io.jnet.common.exception.NotFitProtocolException;
import io.jnet.common.streamprocessor.StreamProcessor;
import io.jnet.common.util.ProcesserUtil;

public class SimpleReadHandler implements ReadHandler
{
    private final StreamProcessor[]         processors;
    private final FrameDecodec              frameDecodec;
    private final AsynchronousSocketChannel socketChannel;
    private final ByteBuf<?>                ioBuf;
    private final ClientChannelContext      clientChannelContext;
    private final ClientListener            clientListener;
    private volatile boolean                readPending = false;
    
    public SimpleReadHandler(ClientListener clientListener, ClientChannelContext clientChannelContext, StreamProcessor[] processors, ByteBuf<?> ioBuf, FrameDecodec frameDecodec, AsynchronousSocketChannel socketChannel)
    {
        this.clientListener = clientListener;
        this.frameDecodec = frameDecodec;
        this.socketChannel = socketChannel;
        this.ioBuf = ioBuf;
        this.processors = processors;
        this.clientChannelContext = clientChannelContext;
    }
    
    @Override
    public void completed(Integer read, Void attachment)
    {
        if (read == -1)
        {
            catchThrowable(EndOfStreamException.instance, clientChannelContext);
            return;
        }
        ioBuf.addWriteIndex(read);
        try
        {
            decodecAndProcess();
            socketChannel.read(getWriteBuffer(), null, this);
        }
        catch (Throwable e)
        {
            catchThrowable(e, clientChannelContext);
        }
    }
    
    @Override
    public void failed(Throwable exc, Void attachment)
    {
        catchThrowable(exc, clientChannelContext);
    }
    
    public void decodecAndProcess() throws Throwable
    {
        while (true)
        {
            DecodeResult decodeResult = frameDecodec.decodec(ioBuf);
            switch (decodeResult.getType())
            {
                case LESS_THAN_PROTOCOL:
                    return;
                case BUF_NOT_ENOUGH:
                    ioBuf.compact().ensureCapacity(decodeResult.getNeed());
                    return;
                case NOT_FIT_PROTOCOL:
                    catchThrowable(NotFitProtocolException.instance, clientChannelContext);
                    return;
                case NORMAL:
                    ByteBuf<?> packet = decodeResult.getBuf();
                    ProcesserUtil.process(clientChannelContext, processors, packet, 0);
            }
        }
    }
    
    /**
     * 将iobuf的内容进行压缩，返回一个处于可写状态的ByteBuffer
     * 
     * @return
     */
    private ByteBuffer getWriteBuffer()
    {
        ByteBuffer ioBuffer = ioBuf.nioBuffer();
        ioBuffer.position(ioBuffer.limit()).limit(ioBuffer.capacity());
        return ioBuffer;
    }
    
    private void catchThrowable(Throwable e, ClientChannelContext context)
    {
        clientListener.catchException(e, context);
    }
    
    public void registerRead()
    {
        if (readPending)
        {
            throw new UnsupportedOperationException();
        }
        readPending = true;
        try
        {
            socketChannel.read(getWriteBuffer(), null, this);
        }
        catch (Exception e)
        {
            clientListener.catchException(e, clientChannelContext);
        }
    }
}
